/**
* Author: Drew Goldberg
* CSCI 5828: HW3
* March 1, 2012
*/
package goldberg.drew.twitter;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public class ConcurrentTwitterV3 extends Twitter{
	final static ExecutorService executorPool = Executors.newFixedThreadPool(9);
	final List<Callable<Object>> userPartition = new ArrayList<Callable<Object>>();	
	final static String topDirectory = "usersConcurrent/";
	/**
	* Empty Constructor
	*/
	private ConcurrentTwitterV3() {
	}
	
	/**
	* Creates the directory hierarchy and textfile for each user	
	* usersConcurrent/userName/userName.txt
	* intializes the threadPool executorPool
	* call the getAllUsersTweets() which will get all the 3200 for each of the 9 users and write it to a file
	*/
	private void init() {

		ConcurrentTwitterV3.createUsers(ConcurrentTwitterV3.topDirectory);
		getAllUsersTweets();	
	}
	
	/**
	* This creates a new ConcurrentTwitterV3 object which call the init function.	
	*/
	public static ConcurrentTwitterV3 create() {
		final ConcurrentTwitterV3 concurrentTwitterV3 = new ConcurrentTwitterV3();
			concurrentTwitterV3.init();
		return concurrentTwitterV3;
	}

	/** Get all the users tweets. The main ExecutorService thread pool: executorPool
	* breaks up getting each users tweets into individual partitions.   
  	* @return void 
  	*/
	public void getAllUsersTweets() {
		final long startTime = System.nanoTime();
		for(String key : users.keySet()) {
			final String userKey = key;
			final File file = users.get(key);
			userPartition.add(new Callable<Object>() { 
				public Object call() {					
					getUserTweets(userKey, file.toString());
					return null;
				}	
			});
		}
		
		try{
			executorPool.invokeAll(userPartition, 1000, TimeUnit.SECONDS);	
		}
		catch(InterruptedException e) {}
		
		executorPool.shutdown();	
		try{
			executorPool.awaitTermination(1000, TimeUnit.SECONDS);
		}
		catch(InterruptedException e) {}
		final long finishTime = System.nanoTime();
		if(executorPool.isTerminated())
			System.out.println("If all the tasks have been completed properly.\nTime take in seconds " + ((finishTime - startTime) / 1.0e9));
		else{
			System.out.println("This was printed because the executor pool threads haven't completed yet.\nTime take in seconds " + (finishTime - startTime / 1.0e9));
		}		
					
	}

	/** request 16 pages of 200 tweets for the userName argument, and written/appended to file argument
  	* @param userName	twitter account userName
  	* @param file		The file that the twitter request will be written to.
  	* @return void 
  	*/
	public void getUserTweets(final String userName, final String file) {
		final ExecutorService pagePool = Executors.newFixedThreadPool(16);
		final List<Callable<Object>> pagePartition = new ArrayList<Callable<Object>>();	
		for(int i = 0; i < 16; i++) {
			final int pageNumber = i+1;
			pagePartition.add(new Callable<Object> (){
				public Object call() {									
					try {	
						System.out.println("Retreiving User " + userName + " page " + pageNumber);				
						System.out.println("Writing to the file " + file + " for page request " + pageNumber);						
						BufferedInputStream in = new BufferedInputStream(new URL("http://api.twitter.com/1/statuses/user_timeline.xml?id="+userName+"&count=200&page="+pageNumber).openStream());
						
						FileOutputStream fos = new FileOutputStream(file, true);
						BufferedOutputStream bout = new BufferedOutputStream(fos, 1024);
						byte[] data = new byte[1024];
						int x = 0;						
						while((x=in.read(data,0,1024))>0){
							bout.write(data, 0, x);
						}
						in.close();						
						bout.close();
											
					}
					catch (IOException e) {
						System.out.println("Failed at userName " + userName + " page " + pageNumber);
						e.printStackTrace();
					}	
					return null;			
				}
				
			});
		}

		try{
			pagePool.invokeAll(pagePartition, 1000, TimeUnit.SECONDS);	
		}
		catch(InterruptedException e) {}
		
		pagePool.shutdown();
		try{
			pagePool.awaitTermination(1000, TimeUnit.SECONDS);		
		}
		catch(InterruptedException e) {
			
		}		
	 }

	public static void main(String[] args) {
		ConcurrentTwitterV3 ct = new ConcurrentTwitterV3();		
		ct.create();
	}

}
